package com.feidee.data.report.util;

import com.feidee.data.report.constant.Constant;
import com.feidee.data.report.dao.IHiveDao;
import com.feidee.data.report.dao.IImpalaDao;
import com.feidee.data.report.dao.IMysqlDao;
import com.feidee.data.report.dao.IReportAndSourceStatusDao;
import com.feidee.data.report.dao.ISnappyDataDao;
import com.feidee.data.report.dao.ISparkDao;
import com.feidee.data.report.exception.HiveSqlExecException;
import com.feidee.data.report.exception.MysqlSqlExecException;
import com.feidee.data.report.model.ReportConfiguration;
import com.feidee.data.report.model.ReportLogModel;
import com.feidee.data.report.model.ReportParam;
import com.feidee.data.report.util.mail.RMailSender;
import com.feidee.data.report.util.threadutil.ThreadPoolUtil;
import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.feidee.fdhadoop.logCollect.LogCollectService;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;

public class QueryThread implements Callable<Integer> {

    private static Logger logger = Logger.getLogger(QueryThread.class);
    private PriorityBlockingQueue<ArrayList<ReportConfiguration>> queryStr;
    private Map<String, ArrayList<ReportConfiguration>> dependencyMap;
    private IHiveDao hiveDao;
    private IImpalaDao impalaDao;
    private IMysqlDao mysqlDao;
    private ISparkDao sparkDao;
    private ISnappyDataDao snappyDataDao;
    private String isCRFlag; // 判断是否为自定义报表
    private String returnUrl; // 自定义报表返回给前端的URL
    private String applyId; // 自定义报表任务ID
    private String history; // 自定义报表执行历史sql标识
    private String admin; //是否为管理员,用于控制是否需要添加运行标识
    private IReportAndSourceStatusDao reportAndSourceStatusDao;  //报表运行状态操作句柄
    private String forceExec;  //是否检验源数据表和依赖报表
    private String groupName;
    private ReportParam rp = null;

    public QueryThread(PriorityBlockingQueue<ArrayList<ReportConfiguration>> queryStr, Map<String, ArrayList<ReportConfiguration>> dependencyMap,
                       IHiveDao hiveDao, IImpalaDao impalaDao, IMysqlDao mysqlDao, ISparkDao sparkDao, String isCRFlag,
                       IReportAndSourceStatusDao reportAndSourceStatusDao, String groupName) {
        this.queryStr = queryStr;
        this.dependencyMap = dependencyMap;
        this.hiveDao = hiveDao;
        this.impalaDao = impalaDao;
        this.mysqlDao = mysqlDao;
        this.sparkDao = sparkDao;
        this.isCRFlag = isCRFlag;
        this.history = "0";
        this.admin = "0";
        this.reportAndSourceStatusDao = reportAndSourceStatusDao;
        this.forceExec = "0";
        this.groupName = groupName;
    }

    public QueryThread(PriorityBlockingQueue<ArrayList<ReportConfiguration>> queryStr, Map<String, ArrayList<ReportConfiguration>> dependencyMap,
                       IHiveDao hiveDao, IImpalaDao impalaDao, IMysqlDao mysqlDao, ISparkDao sparkDao, ISnappyDataDao snappyDataDao,
                       String returnUrl, ReportParam rp, IReportAndSourceStatusDao reportAndSourceStatusDao, String groupName) {
        this.queryStr = queryStr;
        this.dependencyMap = dependencyMap;
        this.hiveDao = hiveDao;
        this.impalaDao = impalaDao;
        this.mysqlDao = mysqlDao;
        this.sparkDao = sparkDao;
        this.snappyDataDao = snappyDataDao;
        this.isCRFlag = rp.getIsCustom();
        this.returnUrl = returnUrl;
        this.applyId = rp.getApplyId();
        this.history = rp.getHistory();
        this.admin = rp.getAdmin();
        this.reportAndSourceStatusDao = reportAndSourceStatusDao;
        this.forceExec = rp.getForceExec();
        this.groupName = groupName;
        this.rp = rp;
    }

    @Override
    public Integer call() throws InterruptedException {

		/* 记录队列中是否还有未执行的报表 */
        boolean isRunning = true;

        try {
            boolean isRunAll = false;
            if (rp != null && "all".equals(rp.getKpi())) {
                isRunAll = true;
            }
            while (isRunning) {
                ArrayList<ReportConfiguration> list = queryStr.poll(300, TimeUnit.SECONDS);
                if (null != list) {
                    int sqlSize = list.size();
                    int reportRes = 0; // 判断该报表是否正常执行
                    long startTimestamp = System.currentTimeMillis();

                    // 报表执行前操作
                    String reportKey = list.get(0).getReport_key();
                    String priority = ToolUtils.getPriorityLevel(list.get(0).getTablePriority());
                    logger.info("编辑发布之后的报表：" + reportKey + " 的优先级为:" + priority);
                    if (beforeExecOperation(list, reportKey, priority, startTimestamp)) {
                        continue;
                    }
                    if (isRunAll) {
                        logger.info("####### 运行全部报表线程 " + Thread.currentThread().getId()
                                + " 开始运行报表：" + reportKey + " #######");
                    }
                    //执行报表
                    ReportConfiguration reportConfiguration = null;
                    String errorMess = "";
                    boolean isRuntimeOccurException = false;  //2020-07-02
                    for (int i = 0; i < sqlSize; i++) {
                        reportConfiguration = list.get(i);
                        // 增加重试机制，如果是spark引擎执行失败切换成hive引擎重试执行一次
                        int j = 0;
                        for (; j < 2; j++) {
                            try {
                                executeHqlSql(reportConfiguration);
                                break;
                            } catch (Exception e) {
                                logger.info("报表:" + reportConfiguration.getReport_key() + "执行出错,错误信息:" + e.getMessage());
                                if ("hive".equals(reportConfiguration.getEngine()) && j == 0 && e.getMessage().contains("code 3")) {
                                   /* logger.info("报表: " + reportConfiguration.getReport_key() + " 第: " + reportConfiguration.getReport_index() +
                                            " 条SQL spark执行失败，切换成hive重新执行!");*/
                                    logger.info("报表" + reportConfiguration.getReport_key() + " 第: " + reportConfiguration.getReport_index()
                                            + "条SQL执行出现code 3错误,正在重新执行...");
                                    //reportConfiguration.setEngine("hive");
                                    //加入重跑标识
                                    RunFlagController.reportIndexSQLIsRestartMap.put(reportConfiguration.getReport_key()
                                            +"_"+reportConfiguration.getReport_index(), "true");
                                } else {
                                    isRuntimeOccurException = true;
                                    reportRes = 1;
                                    reportExecuteFailedResponse("Index:" + (i + 1) + ",Msg:" + e.getMessage());
                                    //记录执行出错报表index
                                    if (reportConfiguration != null) {
                                        errorMess += "Index " + reportConfiguration.getReport_index() + " Error Mess:" + e.getMessage() + "<br/>&nbsp;";
                                    }
                                    break;
                                }
                            }
                        }
                        if (isRuntimeOccurException) {
                            logger.info("很遗憾，报表" + reportConfiguration.getReport_key() + "中途执行出错，提前中止");
                            break;
                        }
                    }
                    if (!errorMess.equals("")) {
                        errorMess = errorMess.substring(0, errorMess.length() - 11);
                    }
                    //报表执行完成后操作
                    afterExecOperation(reportRes, reportKey, priority, startTimestamp, errorMess, reportConfiguration);
                    if (isRunAll) {
                        logger.info("####### 运行全部报表线程 " + Thread.currentThread().getId()
                                + " 运行报表：" + reportKey + " 结束#######");
                    }
                } else if (dependencyMap != null && dependencyMap.size() > 0) {
                    if (isRunAll) {
                        logger.info("####### 运行全部报表线程 " + Thread.currentThread().getId()
                                + " 获取不到可运行报表，尝试唤醒等待报表 #######");
                    }
                    checkDataResourcesAndEnqueue();
                } else {
                    isRunning = false; /* 队列为空,结束该线程 */
                    if (isRunAll) {
                        logger.info("####### 运行全部报表线程 " + Thread.currentThread().getId()
                                + " 获取不到可运行报表，结束运行 #######");
                    }
                }
            }
        } finally {
            ThreadPoolUtil.addThreadSuccessCnt(groupName);
        }

        logger.info("####### Thread " + Thread.currentThread().getId() + " Run Success! Exit Normally! #######");
        return 0;
    }

    private synchronized void checkDataResourcesAndEnqueue() {
        if (dependencyMap != null && dependencyMap.size() > 0) {
            int waitOvertimeNum = 0;
            int dataSourcesFailedNum = 0;
            int dataSourcesNoStatusNum = 0;

            Iterator<Map.Entry<String, ArrayList<ReportConfiguration>>> iterator = dependencyMap.entrySet().iterator();
            while (iterator.hasNext()) {
                try {
                    Map.Entry<String, ArrayList<ReportConfiguration>> next = iterator.next();
                    ReportConfiguration reportConfiguration = next.getValue().get(0);
                    String wakeupReportKey = next.getKey();
                    String dataResources = reportConfiguration.getDataResources();
                    String isScheduler = reportConfiguration.getIsScheduler();
                    long enqueueTime = reportConfiguration.getEnqueueTime();
                    String sourceTblData = reportAndSourceStatusDao.validSourceTblData(forceExec, wakeupReportKey, dataResources,
                            isScheduler, enqueueTime, false);
                    int dataSourcesRes = Integer.valueOf(sourceTblData.split("€", -1)[0]);
                    if (dataSourcesRes != -2) {
                        String validReportDependency = reportAndSourceStatusDao.validReportDependency(forceExec, wakeupReportKey,
                                reportConfiguration.getReportDependency(), enqueueTime, false);
                        int res = Integer.valueOf(validReportDependency.split("€", -1)[0]);
                        if (res != -2) {
                            queryStr.offer(next.getValue());
                            iterator.remove();
                            logger.info("报表: " + wakeupReportKey + " 数据源校验完全通过,重新加入队列成功!");
                        }
                        if (dataSourcesRes == -1) {
                            // 等待超时
                            waitOvertimeNum++;
                        } else if (dataSourcesRes == -3) {
                            // 源表失败
                            dataSourcesFailedNum++;
                        }
                    } else {
                        // 等待数据源
                        dataSourcesNoStatusNum++;
                    }
                } catch (Exception e) {
                    logger.error("校验等待map异常！", e);
                }
            }
            if (rp != null && "all".equals(rp.getKpi())) {
                logger.info("运行全部报表数据源校验情况（等待超时的报表：" + waitOvertimeNum
                        + " 个，源表失败的报表：" + dataSourcesFailedNum + " 个，未能获取数据源状态的报表：" + dataSourcesNoStatusNum + " 个！）");
            }
        }
    }

    /**
     * 报表执行前操作
     *
     * @param list
     * @param reportKey
     * @param priority
     * @param startTimestamp
     * @return
     * @throws InterruptedException
     */
    private boolean beforeExecOperation(ArrayList<ReportConfiguration> list, String reportKey, String priority, long startTimestamp) {

        ReportConfiguration reportConfiguration = list.get(0);
        String isScheduler = reportConfiguration.getIsScheduler();
        int res;
        String errorMsg = "";

        String dataResources = reportConfiguration.getDataResources();
        String sourceTblData = reportAndSourceStatusDao.validSourceTblData(forceExec, reportKey, dataResources, isScheduler,
                reportConfiguration.getEnqueueTime(), true);
        String[] sourceStatusArray = sourceTblData.split("€", -1);
        res = Integer.valueOf(sourceStatusArray[0]);
        if (0 == res) {
            logger.info("报表 " + reportKey + " 依赖源数据表特性校验通过!");
        } else if (-1 == res) {
            logger.info("报表 " + reportKey + " 依赖源数据表特性校验不通过,等待时间达到最大值!");
            errorMsg = "依赖源数据等待时间达到最大限度" + sourceStatusArray[1];
            //更改报表状态为执行等待时间超过最大等待限度
            reportAndSourceStatusDao.changeReportRunStatus(reportKey, Constant.REPORT_DATARESOURCES_WAIT_OVERTIME);
            // 更改报表执行失败时间用于前端展示报表数据最后更新时间
            reportAndSourceStatusDao.updateReportRunTime(reportKey, Constant.REPORT_DATARESOURCES_WAIT_OVERTIME,
                    priority, startTimestamp, errorMsg);
            reportExecuteFailedResponse("报表依赖源数据等待时间达到最大限度,放弃执行");
            //发送报表依赖等待时间达到最大限度预警邮件
            ThreadPoolUtil.addExecExceptionReport(groupName,
                    reportKey + "€" + errorMsg + "€" +
                            TimeUtil.getNow(Constant.LOG_TIME_FORMAT) + "€" + priority, 0L);
            logger.info("**********发送报表 " + reportKey + " 依赖源数据等待时间达到最大限度预警邮件**********");
            return true;
        } else if (-3 == res || -4 == res) {
            logger.info("报表 " + reportKey + " 依赖源数据表同步失败,放弃执行...");
            errorMsg = "源数据同步失败" + sourceStatusArray[1];
            reportAndSourceStatusDao.changeReportRunStatus(reportKey, Constant.REPORT_DATARESOURCES_EXECUTE_FAILED);
            reportExecuteFailedResponse("报表依赖源数据表同步失败");
            // 更改报表执行失败时间用于前端展示报表数据最后更新时间
            reportAndSourceStatusDao.updateReportRunTime(reportKey, Constant.REPORT_DATARESOURCES_EXECUTE_FAILED,
                    priority, startTimestamp, errorMsg);
            //发送源数据校验失败预警邮件
            ThreadPoolUtil.addExecExceptionReport(groupName,
                    reportKey + "€" + errorMsg + "€" +
                            TimeUtil.getNow(Constant.LOG_TIME_FORMAT) + "€" + priority, 0L);
            logger.info("******************发送报表 " + reportKey + " 源数据校验失败预警邮件******************");
            return true;
        } else {
            logger.info("报表 " + reportKey + " 源数据特性校验暂时不通过,继续等待下一次执行...");
            //  将依赖检查不通过的报表加入等待map中
            addToDependencyMap(list);
            return true;
        }

        // 检查报表依赖是否完成
        String validReportDependency = reportAndSourceStatusDao.validReportDependency(forceExec, reportKey,
                reportConfiguration.getReportDependency(), reportConfiguration.getEnqueueTime(), true);
        String[] reportDependencyArray = validReportDependency.split("€", -1);
        res = Integer.valueOf(reportDependencyArray[0]);
        if (0 == res) {
            logger.info("报表 " + reportKey + " 依赖其他报表特性校验通过!");
        } else if (-1 == res) {
            logger.info("报表 " + reportKey + " 依赖其他报表特性校验不通过,并且该报表等待时间达到最大值["
                    + Constant.REPORT_WAIT_MAX_TIME + "]!");
            errorMsg = "依赖其他报表等待时间达到最大限度" + reportDependencyArray[1];
            //更改报表状态为执行等待时间超过最大等待限度
            reportAndSourceStatusDao.changeReportRunStatus(reportKey, Constant.REPORT_DEPENDENCY_WAIT_OVERTIME);
            // 更改报表执行失败时间用于前端展示报表数据最后更新时间
            reportAndSourceStatusDao.updateReportRunTime(reportKey, Constant.REPORT_DEPENDENCY_WAIT_OVERTIME,
                    priority, startTimestamp, errorMsg);
            reportExecuteFailedResponse("报表依赖其他报表等待时间达到最大限度,放弃执行");
            //发送报表依赖等待时间达到最大限度预警邮件
            ThreadPoolUtil.addExecExceptionReport(groupName,
                    reportKey + "€" + errorMsg + "€" +
                            TimeUtil.getNow(Constant.LOG_TIME_FORMAT) + "€" + priority, 0L);
            logger.info("**********发送报表 " + reportKey + " 依赖其他报表等待时间达到最大限度预警邮件**********");
            return true;
        } else if (-3 == res) {
            logger.info("报表 " + reportKey + " 依赖其他报表特性校验不通过,依赖报表中有部分报表执行失败!");
            errorMsg = "依赖部分依赖报表执行失败" + reportDependencyArray[1];
            reportAndSourceStatusDao.changeReportRunStatus(reportKey, Constant.REPORT_DEPENDENCY_EXECUTE_FAILED);
            // 更改报表执行失败时间用于前端展示报表数据最后更新时间
            reportAndSourceStatusDao.updateReportRunTime(reportKey, Constant.REPORT_DEPENDENCY_EXECUTE_FAILED,
                    priority, startTimestamp, errorMsg);
            reportExecuteFailedResponse("报表依赖部分报表执行失败,放弃执行");
            //发送报表依赖部分依赖报表执行失败预警邮件
            ThreadPoolUtil.addExecExceptionReport(groupName,
                    reportKey + "€" + errorMsg + "€" + TimeUtil.getNow(Constant.LOG_TIME_FORMAT)
                            + "€" + priority, 0L);
            logger.info("*********发送报表 " + reportKey + " 依赖部分依赖报表执行失败预警邮件*********");
            return true;
        } else if (-4 == res) {
            logger.info("报表 " + reportKey + " 依赖其他报表特性校验不通过,依赖报表中有部分报表执行等待超时,未执行...");
            errorMsg = "部分依赖的报表执行等待超时" + reportDependencyArray[1];
            reportAndSourceStatusDao.changeReportRunStatus(reportKey, Constant.REPORT_DEPENDENCY_EXECUTE_FAILED);
            // 更改报表执行失败时间用于前端展示报表数据最后更新时间
            reportAndSourceStatusDao.updateReportRunTime(reportKey, Constant.REPORT_DEPENDENCY_EXECUTE_FAILED,
                    priority, startTimestamp, errorMsg);
            reportExecuteFailedResponse("报表依赖部分报表执行等待超时,放弃执行");
            //发送报表依赖部分依赖报表执行等待超时预警邮件
            ThreadPoolUtil.addExecExceptionReport(groupName,
                    reportKey + "€" + errorMsg + "€" + TimeUtil.getNow(Constant.LOG_TIME_FORMAT)
                            + "€" + priority, 0L);
            logger.info("*********发送报表 " + reportKey + " 部分依赖的报表执行等待超时预警邮件*********");
            return true;
        } else {
            logger.info("报表 " + reportKey + " 依赖其他报表特性校验暂时不通过,继续等待下一次执行...");
            //  将依赖检查不通过的报表加入等待map中
            addToDependencyMap(list);
            return true;
        }

        // 停止其他运行标识并添加运行标识
        if ("0".equals(admin)) {
            RunFlagController.addRunFlag(reportKey, Thread.currentThread().getId());
            logger.info("添加报表标识 reportKey:" + reportKey + ",ThreadId:" + Thread.currentThread().getId()
                    + ",当前正在运行的报表：" + RunFlagController.getRunFlagCacheStr());
        } else {
            logger.info("管理员运行报表 reportKey:" + reportKey + ",ThreadId:" + Thread.currentThread().getId()
                    + ",无需添加运行标识,当前正在运行的报表：" + RunFlagController.getRunFlagCacheStr());
        }
        //更改报表状态为正在执行
        reportAndSourceStatusDao.changeReportRunStatus(reportKey, Constant.REPORT_DEPENDENCY_EXECUTING);
        reportAndSourceStatusDao.updateReportRunTime(reportKey, Constant.REPORT_DEPENDENCY_EXECUTING, priority,
                startTimestamp, errorMsg);

        return false;
    }

    private void addToDependencyMap(ArrayList<ReportConfiguration> list) {
        if (dependencyMap != null) {
            String report_key = list.get(0).getReport_key();
            dependencyMap.put(report_key, list);
            logger.info("报表 " + report_key + " 加入依赖等待map成功！");
        }
    }

    /*
     * 执行一组表报里的一个HqlSql
     */
    private void executeHqlSql(ReportConfiguration configuration) throws Exception {

        String engine = "hive";
        if (ToolUtils.isNotNull(configuration.getEngine())) {
            engine = configuration.getEngine();
        }
        ReportLogModel logs = new ReportLogModel();
        logs.setReport_key(configuration.getReport_key());
        logs.setReport_index(configuration.getReport_index());
        logs.setStatus(Constant.LOG_START);
        logs.setTs(TimeUtil.getNow(Constant.LOG_TIME_FORMAT));
        logs.setMsg("");
        logs.setWriteCnt("0");
        String enqueueTimeStr = ToolUtils.getTimeStr(configuration.getEnqueueTime());
        logs.setEnqueueTime(enqueueTimeStr);
        logs.setRunHistory(configuration.getRunHistory());
        long curTimestamp = System.currentTimeMillis();
        String queueTime = ToolUtils.getCostTimeStr(curTimestamp - configuration.getEnqueueTime());
        logs.setCostTime(queueTime);

        LogCollectService.sendLog(CustomConfiguration.getString(Constant.LOG_APT),
                CustomConfiguration.getString(Constant.LOG_BUS), logs);
        configuration.setEnqueueTime(curTimestamp); //将加入队列时间设置为开始执行时间，方便执行结束时统计执行耗时

        logger.info("####### THREAD " + Thread.currentThread().getId() + " --> Run KPI: "
                + configuration.getReport_key() + " _" + configuration.getReport_index()
                + " ,RunPriority:" + configuration.getRunPriority() + " ,Enqueue Time:"
                + enqueueTimeStr + " ,RunHistorySql:" + configuration.getRunHistory()
                + " ,QueueTime:" + queueTime + " #######");

        // 执行报表逻辑
        try {
            if ("hive".equalsIgnoreCase(engine)) {
                hiveDao.query(configuration, this.rp);
            } else if ("impala".equalsIgnoreCase(engine)) {
                impalaDao.query(configuration, this.rp);
            } else if ("mysql".equalsIgnoreCase(engine)) {
                mysqlDao.query(configuration, this.rp);
            } else if ("spark".equalsIgnoreCase(engine)) {
                sparkDao.query(configuration, this.rp);
            } else if ("snappydata".equalsIgnoreCase(engine)) {
                snappyDataDao.query(configuration, this.rp);
            }
        } catch (HiveSqlExecException e) {
            logger.error("####### THREAD " + Thread.currentThread().getId() + " Execute SQL: "
                    + configuration.getReport_key() + " _" + configuration.getReport_index() + " Exception!");
            throw new Exception("Hive_SQL执行失败," + e.getMessage());
        } catch (MysqlSqlExecException e) {
            logger.error("####### THREAD " + Thread.currentThread().getId() + " Execute SQL: "
                    + configuration.getReport_key() + " _" + configuration.getReport_index() + " Exception!");
            throw new Exception("Mysql_SQL执行失败," + e.getMessage());
        } catch (Exception e) {
            logger.error("####### THREAD " + Thread.currentThread().getId() + " Execute SQL: "
                    + configuration.getReport_key() + " _" + configuration.getReport_index() + " Exception!");
            throw new Exception("报表执行未知异常," + e.getMessage());
        }
    }

    /**
     * 报表执行完成后操作
     *
     * @param reportRes
     * @param reportKey
     * @param priority
     * @param startTimestamp
     * @param errorMess
     * @param reportConfiguration
     */
    private void afterExecOperation(int reportRes, String reportKey, String priority, long startTimestamp,
                                    String errorMess, ReportConfiguration reportConfiguration) {

        //检测是否需要返回报表执行成功状态信息
        if (0 == reportRes && "1".equals(isCRFlag)) {
            //如果运行的是历史SQL而且没有中途被停止，则发送执行成功的信息，否则不发送[因为后期增加了历史SQL执行成功的标识]
            if ("1".equals(history) &&
                    RunFlagController.checkRunFlag(reportKey, Thread.currentThread().getId())) {
                ResultResponseUtil.resultResponse(returnUrl, applyId, "5", "自定义报表历史SQL执行成功"); //执行成功
            } else if ("0".equals(history)) {
                ResultResponseUtil.resultResponse(returnUrl, applyId, "2", "自定义报表实时SQL执行成功"); //执行成功
            }
        }

        //更新报表执行状态流程如下：
        //(1)如果报表中途未被停止,执行成功，则将状态更新为执行成功，否则状态更新为执行失败；
        //(2)如果
        // 报表中途被停止了，并且是手动停止[自定义报表有手动停止历史SQL执行的功能],则更新报表状态为执行失败；
        //(3)如果报表中途被停止了，并且不是手动停止，是被用户另外一次提交导致本次结束的情况，则不更新状态，等待另外一次提交执行结果更新该状态；
        if (RunFlagController.checkRunFlag(reportKey, Thread.currentThread().getId())) {
            if (0 == reportRes) {
                //记录执行成功报表信息
                ThreadPoolUtil.addExecSuccessReport(groupName, reportKey + ";执行成功;"
                        + TimeUtil.getNow(Constant.LOG_TIME_FORMAT) + ";" + priority, System.currentTimeMillis() - startTimestamp);
                reportAndSourceStatusDao.changeReportRunStatus(reportKey, Constant.REPORT_DEPENDENCY_EXECUTE_SUCCESS);
                // 更改报表执行成功时间用于前端展示报表数据最后更新时间
                reportAndSourceStatusDao.updateReportRunTime(reportKey, Constant.REPORT_DEPENDENCY_EXECUTE_SUCCESS,
                        priority, startTimestamp, errorMess);
            } else {
                reportAndSourceStatusDao.changeReportRunStatus(reportKey, Constant.REPORT_DEPENDENCY_EXECUTE_FAILED);
                // 更改报表执行失败时间用于前端展示报表数据最后更新时间
                reportAndSourceStatusDao.updateReportRunTime(reportKey, Constant.REPORT_DEPENDENCY_EXECUTE_FAILED,
                        priority, startTimestamp, errorMess);
                //如果中途未停止,报表执行失败,发送预警邮件
                ThreadPoolUtil.addExecExceptionReport(groupName, reportKey + "€" + errorMess + "€"
                        + TimeUtil.getNow(Constant.LOG_TIME_FORMAT) + "€" + priority, System.currentTimeMillis() - startTimestamp);
                logger.info("*********报表 " + reportKey + " 执行失败而且中途未停止,发送预警邮件*********");
            }
        } else if (RunFlagController.checkStopRunFlag(reportKey, Thread.currentThread().getId())) {
            reportAndSourceStatusDao.changeReportRunStatus(reportKey, Constant.REPORT_DEPENDENCY_EXECUTE_FAILED);
            logger.info("*********报表 " + reportKey + " 执行中途停止,将状态更新为执行失败*********");
            // 更改报表执行失败时间用于前端展示报表数据最后更新时间
            reportAndSourceStatusDao.updateReportRunTime(reportKey, Constant.REPORT_DEPENDENCY_EXECUTE_FAILED,
                    priority, startTimestamp, errorMess);
        }

        //清楚报表运行标识
        if ("0".equals(admin)) {
            RunFlagController.clearRunFlag(reportKey, Thread.currentThread().getId());
            logger.info("清除报表标识reportKey:" + reportKey + ",ThreadId:" + Thread.currentThread().getId()
                    + ",当前正在运行的报表：" + RunFlagController.getRunFlagCacheStr());
        } else {
            logger.info("管理员运行报表 reportKey:" + reportKey + ",ThreadId:" + Thread.currentThread().getId()
                    + ",当前正在运行的报表：" + RunFlagController.getRunFlagCacheStr());
        }

        //发送报表订阅邮件(自定义报表历史sql执行完不发送邮件)
        if ("0".equals(history)) {
            if (0 == reportRes) {
                RMailSender.sendReportMail(reportConfiguration.getReport_key(), 1);
            } else {
                RMailSender.sendReportMail(reportConfiguration.getReport_key(), 0);
            }
        }

        //  检查该报表是都被其他报表依赖，若依赖该报表的报表在依赖等待map中则将其重新入列
        checkDataResourcesAndEnqueue();

    }

    /**
     * 自定义报表执行结果返回
     *
     * @param msg
     */
    private void reportExecuteFailedResponse(String msg) {
        if ("1".equals(isCRFlag)) {
            if ("1".equals(history)) {
                ResultResponseUtil.resultResponse(returnUrl, applyId, "4", "[历史数据]" + msg); // 历史sql执行失败
            } else {
                ResultResponseUtil.resultResponse(returnUrl, applyId, "3", msg); // 实时SQL执行失败
            }
        }
    }

}
